package org.lql.chaos.rocketmq.service.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: liangqinglong
 * @date: 2025-08-17 00:40
 * @description: 消费者监控服务
 **/
@Slf4j
@Service
public class ConsumerMonitorService {

    private final ConcurrentHashMap<String, DefaultMQPushConsumer> consumers = new ConcurrentHashMap<>();

    /**
     * 创建并启动一个新的消费者实例
     *
     * @param consumerGroup 消费者组名
     * @param topic         主题
     * @param tag           标签表达式
     * @return 消费者实例
     * @throws MQClientException MQ客户端异常
     */
    public DefaultMQPushConsumer createAndStartConsumer(String consumerGroup, String topic, String tag) 
            throws MQClientException {
        // 如果消费者已存在，直接返回
        if (consumers.containsKey(consumerGroup)) {
            log.warn("消费者组 {} 已存在", consumerGroup);
            return consumers.get(consumerGroup);
        }

        // 创建消费者实例
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        
        // 订阅主题和标签
        consumer.subscribe(topic, tag);
        
        // 启动消费者
        consumer.start();
        
        // 添加到管理列表
        consumers.put(consumerGroup, consumer);
        
        log.info("成功创建并启动消费者组: {}, 订阅主题: {}, 标签: {}", consumerGroup, topic, tag);
        return consumer;
    }

    /**
     * 停止指定的消费者
     *
     * @param consumerGroup 消费者组名
     */
    public void shutdownConsumer(String consumerGroup) {
        DefaultMQPushConsumer consumer = consumers.remove(consumerGroup);
        if (consumer != null) {
            consumer.shutdown();
            log.info("已停止消费者组: {}", consumerGroup);
        } else {
            log.warn("未找到消费者组: {}", consumerGroup);
        }
    }

    /**
     * 应用关闭时停止所有消费者
     */
    @PreDestroy
    public void shutdownAllConsumers() {
        consumers.forEach((group, consumer) -> {
            try {
                consumer.shutdown();
                log.info("已停止消费者组: {}", group);
            } catch (Exception e) {
                log.error("停止消费者组 {} 失败", group, e);
            }
        });
        consumers.clear();
        log.info("所有消费者已停止");
    }

    /**
     * 获取消费者状态信息
     *
     * @return 消费者组列表
     */
    public ConcurrentHashMap<String, DefaultMQPushConsumer> getConsumers() {
        return consumers;
    }
}